//-< STOCKDB.CPP >---------------------------------------------------*--------*
// GigaBASE                  Version 1.0         (c) 1999  GARRET    *     ?  *
// (Post Relational Database Management System)                      *   /\|  *
//                                                                   *  /  \  *
//                          Created:     22-Jan-2012   K.A. Knizhnik * / [] \ *
//                          Last update: 22-Jan-2012   K.A. Knizhnik * GARRET *
//-------------------------------------------------------------------*--------*
// Stock database
//-------------------------------------------------------------------*--------*

#include "stockdb.h"
#include "datetime.h"
#include <errno.h>

const time_t SECONDS_IN_DAY = 24*60*60;
const size_t QUOTES_PER_BLOCK = 256;
const time_t BLOCK_INTERVAL = QUOTES_PER_BLOCK*SECONDS_IN_DAY*7/5;
const size_t IMPORT_FILE_LINE_LIMIT = 256;
const size_t MAX_SYMBOL_LENGTH = 32;
const size_t MAX_FILE_NAME_LENGTH = 256;
const size_t FORMAT_BUFFER_LENGTH = 1024;
const size_t PAGE_POOL_SIZE = 64*1024*1024; // do not need to much for time series because of mostly full scan queries

typedef dbTimeSeriesBlock<Quote>  DailyBlock;
map<string, Processor*> Processor::processors;

REGISTER_TEMPLATE(DailyBlock);
REGISTER(Stock);

string StockDB::format(char const* fmt, ...)
{
    char buf[FORMAT_BUFFER_LENGTH];
    va_list args;
    va_start(args, fmt);
    vsprintf(buf, fmt, args);
    va_end(args);
    return string(buf);
}

char* StockDB::printDate(date_t dt, char* buf, size_t bufSize)
{
    dbDateTime date(dt);
    return date.asString(buf, bufSize, "%Y-%m-%d");
}

bool StockDB::parseDate(date_t& dt, char const* buf)
{
    int y, m, d, n;
    if ((sscanf(buf, "%d-%d-%d%n",  &y, &m, &d, &n) != 3 && sscanf(buf, "%d/%d/%d%n",  &m, &d, &y, &n) != 3) || size_t(n) != strlen(buf)) { 
        return false;
    }
    dbDateTime date(y, m, d);
    dt = date_t(date.asTime_t());
    return dt != date_t(-1);
}

void StockDB::addQuotes(char const* sym, vector<Quote> const& quotes)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    dbQuery query;
    query = "name=",symbol;
    dbCursor<Stock> stocks(dbCursorForUpdate);
    oid_t stockId;
    if (stocks.select(query)) { 
        stockId = stocks.getOid();
    } else { 
        Stock stock;
        stock.name = symbol;
        stock.from = stock.till = date_t(time(NULL));
        stock.nQuotes = 0;
        dbReference<Stock> r = insert(stock);
        stocks.at(r);
        stockId = r.getOid();
    }
    bool descendingOrder = provider->descendingOrder();
    for (size_t i = 0, n = quotes.size(); i < n; i++) { 
        Quote const& quote = quotes[i];
        if (processor.add(stockId, quote, descendingOrder)) { 
            if (stocks->nQuotes == 0 || quote.date < stocks->from) { 
                stocks->from = quote.date;
            }
            if (stocks->nQuotes == 0 || quote.date > stocks->till) { 
                stocks->till = quote.date;
            }
            stocks->nQuotes += 1;
        }
    }
    stocks.update();
}

ErrorCode StockDB::importQuotes(char const* sym, char const* csvFileName, size_t& nImported, bool skipHeader)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    char buf[IMPORT_FILE_LINE_LIMIT];
    FILE* f = fopen(csvFileName, "r");
    if (f == NULL) { 
        error = format("File '%s' not found", csvFileName);;
        return FILE_NOT_FOUND;
    }
    if (skipHeader) { 
        if (fgets(buf, sizeof buf, f) == NULL) { 
            error = "File has not header";
            return FILE_FORMAT_ERROR;
        }
    }
    ErrorCode err = importQuotes(symbol, f, nImported);
    fclose(f);
    return err;    
}

ErrorCode StockDB::importQuotes(char const* sym, FILE* csvData, size_t& nImported)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    int m,d,y;
    Quote quote;
    char buf[IMPORT_FILE_LINE_LIMIT];
    dbQuery query;
    query = "name=",symbol;
    dbCursor<Stock> stocks(dbCursorForUpdate);
    oid_t stockId;
    if (stocks.select(query)) { 
        stockId = stocks.getOid();
    } else { 
        Stock stock;
        stock.name = symbol;
        stock.from = stock.till = date_t(time(NULL));
        stock.nQuotes = 0;
        dbReference<Stock> r = insert(stock);
        stocks.at(r);
        stockId = r.getOid();
    }
    bool descendingOrder = provider->descendingOrder();
    size_t nQuotes = 0;
    while (fgets(buf, sizeof buf, csvData) != NULL) { 
        quote.adjClose = 0;
        if (sscanf(buf, "%d-%d-%d,%f,%f,%f,%f,%d,%f", &y, &m, &d, &quote.open, &quote.high, &quote.low, &quote.close, &quote.volume, &quote.adjClose) < 8
            && sscanf(buf, "%d/%d/%d,%f,%f,%f,%f,%d,%f", &m, &d, &y, &quote.open, &quote.high, &quote.low, &quote.close, &quote.volume, &quote.adjClose) < 8) 
        { 
            error = "Not a valid CVS file";
            return FILE_FORMAT_ERROR;
        }
        dbDateTime date(y, m, d);
        time_t t = date.asTime_t();
        if (t - BLOCK_INTERVAL != date_t(t - BLOCK_INTERVAL)) { 
            break;
        }
        quote.date = date_t(t);
        if (processor.add(stockId, quote, descendingOrder)) { 
            if (nQuotes == 0 || quote.date < stocks->from) { 
                stocks->from = quote.date;
            }
            if (nQuotes == 0 || quote.date > stocks->till) { 
                stocks->till = quote.date;
            }
            nQuotes += 1;
        }
    }
    nImported = nQuotes;
    stocks->nQuotes += nQuotes;
    stocks.update();
    return OK;
}

void StockDB::flush()
{
    db.commit();
}

bool StockDB::backup(char const* filePath, bool compact) 
{ 
    return db.backup(filePath, compact ? dbDatabase::BCK_COMPACTIFY : dbDatabase::BCK_INCREMENTAL);
}

vector<Quote> StockDB::getQuotes(char const* sym, date_t from, date_t till)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    dbQuery query;
    query = "name=",symbol;
    dbCursor<Stock> stocks;
    vector<Quote> quotes;
    if (stocks.select(query)) { 
        quotes.resize((till-from+SECONDS_IN_DAY-1)/SECONDS_IN_DAY);
        quotes.resize(processor.getInterval(stocks.getOid(), from, till, &quotes[0], quotes.size()));
    }
    return quotes;
}
 
ErrorCode StockDB::processQuotes(Processor* proc, char const* sym, date_t from, date_t till, FILE* csvFile)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    dbQuery query;
    query = "name=",symbol;
    dbCursor<Stock> stocks;
    if (stocks.select(query)) { 
        Processor::Callback* callback = proc->getCallback(csvFile);
        processor.select(stocks.getOid(), from, till, callback);
        delete callback;
    }
    return OK;
}

ErrorCode StockDB::processQuotes(char const* procName, char const* symbol, date_t from, date_t till, FILE* csvFile)
{
    Processor* processor = Processor::findProcessor(procName);
    if (processor == NULL) { 
        return NO_PROCESSOR;
    }
    return processQuotes(processor, symbol, from, till, csvFile);
}


ErrorCode StockDB::exportQuotes(char const* sym, date_t from, date_t till, char const* csvFileName, bool writeHeader)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    char const* suffix = strrchr(csvFileName, '.');
    char resultFile[MAX_FILE_NAME_LENGTH];
    if (suffix == NULL) { 
        sprintf(resultFile, "%s.csv", csvFileName);
        csvFileName = resultFile;
    }
    FILE* f = fopen(csvFileName, "w");
    if (f == NULL) { 
        error = format("Failed to create file '%s': %d", csvFileName, errno);
        return FILE_NOT_FOUND;
    }
    if (writeHeader) { 
        if (fprintf(f, "Date,Open,High,Low,Close,Volume,Adj Close\n") < 0) { 
            error = format("Failed to write file: %d", errno);
            return FILE_WRITE_ERROR;
        }
     }

    ErrorCode err = exportQuotes(symbol, from, till, f);
    fclose(f);
    return err;    
}
 
ErrorCode StockDB::exportQuotes(char const* sym, date_t from, date_t till, FILE* csvFile)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    vector<Quote> quotes = getQuotes(symbol, from, till);
    char buf[TIME_BUF_SIZE];
    for (size_t i = 0, n = quotes.size(); i < n; i++) {
        Quote const& quote = quotes[i];
        if (fprintf(csvFile, "%s,%f,%f,%f,%f,%d,%f\n", printDate(quote.date, buf, sizeof buf), quote.open, quote.high, quote.low, quote.close, quote.volume, quote.adjClose) < 0) {
            error = format("Failed to write file: %d", errno);
            return FILE_WRITE_ERROR;
        }
    }
    return OK;
}

vector<string> getSymbols()
{
    dbCursor<Stock> stocks;
    vector<string> symbols;
    if (stocks.select()) { 
        do { 
            symbols.push_back(stocks->name);
        } while (stocks.next());
    }
    return symbols;
}

Stock StockDB::getStock(char const* sym)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    dbQuery query;
    query = "name=",symbol;
    dbCursor<Stock> stocks;
    Stock stock;
    stock.name = symbol;
    if (stocks.select(query)) { 
        stock.from = stocks->from;
        stock.till = stocks->till;
        stock.nQuotes = stocks->nQuotes;
    } else { 
        stock.from = 0;
        stock.till = 0;
        stock.nQuotes = 0;
    }
    return stock;
}

vector<Stock> StockDB::getStocks()
{
    dbCursor<Stock> stocks;
    vector<Stock> result;
    if (stocks.select()) { 
        do { 
            result.push_back(*stocks.get());
        } while (stocks.next());
    }
    return result;
}

int StockDB::removeQuotes(char const* sym, date_t from, date_t till)
{
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    dbQuery query;
    query = "name=",symbol;
    dbCursor<Stock> stocks(dbCursorForUpdate);
    Stock stock;
    stock.name = symbol;
    if (stocks.select(query)) { 
        int nRemoved = processor.remove(stocks.getOid(), from, till);
        stocks->nQuotes -= nRemoved;
        if (stocks->nQuotes == 0) { 
            stocks.remove();
        } else { 
            if (till >= stocks->till) { 
                stocks->till = processor.getLastTime(stocks.getOid());
            }
            if (from <= stocks->from) { 
                stocks->from = processor.getFirstTime(stocks.getOid());
            }
            stocks.update();
        }
        return nRemoved;
    }
    return 0;
}

ErrorCode StockDB::update(bool progress)
{
    dbCursor<Stock> stocks(dbCursorForUpdate); // avoid deadlock
    if (stocks.select()) { 
        do { 
            ErrorCode err = loadSymbol(stocks->name.c_str(), stocks->till + SECONDS_IN_DAY, 0, progress);
            if (err != OK) { 
                return err;
            }
        } while (stocks.next());
    }
    return OK;
}            

ErrorCode StockDB::load(char const* fileWithSymbols, date_t from, date_t till, bool progress)
{
    char symbol[MAX_SYMBOL_LENGTH];
    FILE* f = fopen(fileWithSymbols, "r");
    if (f == NULL) { 
        return FILE_NOT_FOUND;
    }
    while (fscanf(f, "%s", symbol) == 1) { 
        strlower(symbol, symbol);
        if (progress) { 
            printf("Loading symbol %s...\n", symbol);
        }
        ErrorCode err = loadSymbol(symbol, from, till, progress);
        if (err == SYMBOL_NOT_FOUND) { 
            fprintf(stderr, "Symbol %s not found\n", symbol);
        } else if (err != OK) { 
            return err;
        }
    }
    fclose(f);
    return OK;
}

ErrorCode StockDB::loadSymbol(char const* sym, date_t from, date_t till, bool progress)
{
    string file;
    char symbol[MAX_SYMBOL_LENGTH];
    strlower(symbol, sym);
    if (provider == NULL) { 
        error = "No data provider";
        return NO_PROVIDER;
    }
    if (from == 0) { 
        dbQuery query;
        query = "name=",symbol;
        dbCursor<Stock> stocks(dbCursorForUpdate); // avoid deadlock
        if (stocks.select(query)) { 
            time_t last = processor.getLastTime(stocks.getOid());
            if (last != (time_t)-1) {
                from = date_t(last + SECONDS_IN_DAY);
            }
        }
    }
    if (from / SECONDS_IN_DAY >= time(NULL) / SECONDS_IN_DAY) { 
        // nothing to retrieve
        return OK;
    }
    ErrorCode err = provider->download(file, error, symbol, from, till);
    if (err == OK) { 
        size_t nQuotes;
        err = importQuotes(symbol, file.c_str(), nQuotes, true);
        if (progress) { 
            printf("Loaded %d new quotes for symbol %s\n", int(nQuotes), symbol);
        }
        provider->releaseFile(file);
    }
    return err;

}

StockDB::QuoteProcessor::QuoteProcessor(dbDatabase& database)
: dbTimeSeriesProcessor<Quote>(database, QUOTES_PER_BLOCK, QUOTES_PER_BLOCK, BLOCK_INTERVAL) 
{
}

void StockDB::QuoteProcessor::process(Quote const& quote, void* ctx)
{
    ((Processor::Callback*)ctx)->apply(quote);
}


StockDB::StockDB(StockDataProvider* service) 
  : db(dbDatabase::dbAllAccess, PAGE_POOL_SIZE/dbPageSize),
    processor(db), 
    provider(service) 
{
}

ErrorCode StockDB::open(char const* databaseFilePath)
{
    return db.open(databaseFilePath) ? OK : DATABASE_ERROR;
}

void StockDB::close()
{
    db.close();
}

class GroupByMonthProcessor : public Processor
{
    class GroupByMonthCallback : public Callback 
    {
        XQuote agg;
        int year;
        int month;

        void printQuote() { 
            if (agg.date != 0) { 
                fprintf(out, "%d-%d,%f,%f,%f,%f," INT8_FORMAT_PREFIX "d,%f\n", 
                        year, month, agg.open, agg.high, agg.low, agg.close, agg.volume, agg.adjClose);
            }
        }

      public:
        GroupByMonthCallback(FILE* out) : Callback(out) 
        {
            agg.date = 0;
            agg.volume = 0;
        }

        void apply(Quote const& quote) 
        { 
            if (quote.date > agg.date) { 
                printQuote();
                dbDateTime curr(quote.date);
                month = curr.month();
                year = curr.year();
                dbDateTime nextMonth(month == 12 ? year + 1 : year, month == 12 ? 1 : month + 1, 1);
                agg.date = nextMonth.asTime_t();
                agg.open = quote.open;
                agg.high = quote.high;
                agg.low = quote.low;
            } else { 
                if (quote.high > agg.high) {
                    agg.high = quote.high;
                }
                if (quote.low < agg.low) {
                    agg.low = quote.low;
                }
            }
            agg.close = quote.close; 
            agg.adjClose = quote.adjClose; 
            agg.volume += quote.volume;
        }
        
        ~GroupByMonthCallback() { 
            printQuote();
        }
    };

                
  public:
    GroupByMonthProcessor() : Processor("group-by-month") {}

    Callback* getCallback(FILE* file) { 
        return new GroupByMonthCallback(file);
    }
};

class GroupByYearProcessor : public Processor
{
    class GroupByYearCallback : public Callback 
    {
        XQuote agg;
        int year;

        void printQuote() { 
            if (agg.date != 0) { 
                fprintf(out, "%d,%f,%f,%f,%f," INT8_FORMAT_PREFIX "d,%f\n", 
                        year, agg.open, agg.high, agg.low, agg.close, agg.volume, agg.adjClose);
            }
        }

      public:
        GroupByYearCallback(FILE* out) : Callback(out) 
        {
            agg.date = 0;
            agg.volume = 0;
        }

        void apply(Quote const& quote) 
        { 
            if (quote.date > agg.date) { 
                printQuote();
                dbDateTime curr(quote.date);
                year = curr.year();
                dbDateTime nextYear(year + 1, 1, 1);
                agg.date = nextYear.asTime_t();
                agg.open = quote.open;
                agg.high = quote.high;
                agg.low = quote.low;
            } else { 
                if (quote.high > agg.high) {
                    agg.high = quote.high;
                }
                if (quote.low < agg.low) {
                    agg.low = quote.low;
                }
            }
            agg.close = quote.close; 
            agg.adjClose = quote.adjClose; 
            agg.volume += quote.volume;
        }
        
        ~GroupByYearCallback() { 
            printQuote();
        }
    };

                
  public:
    GroupByYearProcessor() : Processor("group-by-year") {}

    Callback* getCallback(FILE* file) { 
        return new GroupByYearCallback(file);
    }
};

class ReturnProcessor : public Processor
{
    class ReturnCallback : public Callback 
    {
        float close;

      public:
        ReturnCallback(FILE* out) : Callback(out) 
        {
            close = 0;
        }

        void apply(Quote const& quote) 
        { 
            if (close != 0) { 
                char buf[TIME_BUF_SIZE];
                fprintf(out, "%s,%f\n", StockDB::printDate(quote.date, buf, sizeof buf), (quote.close - close)/close);
            }
            close = quote.close;
        }
    };

                
  public:
    ReturnProcessor() : Processor("return") {}

    Callback* getCallback(FILE* file) { 
        return new ReturnCallback(file);
    }
};

class MaxProcessor : public Processor
{
    class MaxCallback : public Callback 
    {
        Quote max;

      public:
        MaxCallback(FILE* out) : Callback(out) 
        {
            max.date = 0;
        }

        void apply(Quote const& quote) 
        { 
            if (max.date == 0) { 
                max = quote;
            } else {
                if (max.high < quote.high) { 
                    max.high = quote.high;
                }
                if (max.low < quote.low) { 
                    max.low = quote.low;
                }
                if (max.open < quote.open) { 
                    max.open = quote.open;
                }
                if (max.close < quote.close) { 
                    max.close = quote.close;
                }
                if (max.volume < quote.volume) { 
                    max.volume = quote.volume;
                }
                if (max.adjClose < quote.adjClose) { 
                    max.adjClose = quote.adjClose;
                }
            }
        }
        
        ~MaxCallback() { 
            if (max.date != 0) { 
                fprintf(out, "%f,%f,%f,%f,%d,%f\n", max.open, max.high, max.low, max.close, max.volume, max.adjClose);
            }
        }
    };

                
  public:
    MaxProcessor() : Processor("max") {}

    Callback* getCallback(FILE* file) { 
        return new MaxCallback(file);
    }
};

class MinProcessor : public Processor
{
    class MinCallback : public Callback 
    {
        Quote min;

      public:
        MinCallback(FILE* out) : Callback(out) 
        {
            min.date = 0;
        }

        void apply(Quote const& quote) 
        { 
            if (min.date == 0) { 
                min = quote;
            } else {
                if (min.high > quote.high) { 
                    min.high = quote.high;
                }
                if (min.low > quote.low) { 
                    min.low = quote.low;
                }
                if (min.open > quote.open) { 
                    min.open = quote.open;
                }
                if (min.close > quote.close) { 
                    min.close = quote.close;
                }
                if (min.volume > quote.volume) { 
                    min.volume = quote.volume;
                }
                if (min.adjClose > quote.adjClose) { 
                    min.adjClose = quote.adjClose;
                }
            }
        }
        
        ~MinCallback() { 
            if (min.date != 0) { 
                fprintf(out, "%f,%f,%f,%f,%d,%f\n", min.open, min.high, min.low, min.close, min.volume, min.adjClose);
            }
        }
    };

                
  public:
    MinProcessor() : Processor("min") {}

    Callback* getCallback(FILE* file) { 
        return new MinCallback(file);
    }
};

class AvgProcessor : public Processor
{
    class AvgCallback : public Callback 
    {
        date_t from;
        XQuote avg;
        size_t count;

      public:
        AvgCallback(FILE* out) : Callback(out) 
        {
            count = 0;
            memset(&avg, 0, sizeof avg);
        }

        void apply(Quote const& quote) 
        { 
            if (count == 0) { 
                from = quote.date;
            } else {              
                avg.date = quote.date;
            }
            avg.high += quote.high;
            avg.low += quote.low;
            avg.open += quote.open;
            avg.close += quote.close;
            avg.volume += quote.volume;
            avg.adjClose += quote.adjClose;
            count += 1;
        }
        
        ~AvgCallback() { 
            if (count != 0) { 
                char buf[TIME_BUF_SIZE];
                fprintf(out, "%s,%f,%f,%f,%f," INT8_FORMAT_PREFIX "d,%f\n", StockDB::printDate((avg.date + from)/2, buf, sizeof buf), 
                        avg.open, avg.high, avg.low, avg.close, avg.volume, avg.adjClose);
            }
        }
    };

                
  public:
    AvgProcessor() : Processor("avg") {}

    Callback* getCallback(FILE* file) { 
        return new AvgCallback(file);
    }
};

class CountProcessor : public Processor
{
    class CountCallback : public Callback 
    {
        size_t count;

      public:
        CountCallback(FILE* out) : Callback(out) 
        {
            count = 0;
        }

        void apply(Quote const& quote) 
        { 
            count += 1;
        }
        
        ~CountCallback() { 
            fprintf(out, "%ld\n", count);
        }
    };

                
  public:
    CountProcessor() : Processor("count") {}

    Callback* getCallback(FILE* file) { 
        return new CountCallback(file);
    }
};

static GroupByMonthProcessor groupByMonthProcessor;
static GroupByYearProcessor groupByYearProcessor;
static MaxProcessor maxProcessor;
static MinProcessor minProcessor;
static AvgProcessor avgProcessor;
static CountProcessor countProcessor;
static ReturnProcessor returnProcessor;
